package util;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.serializer.SerializeConfig;
import org.apache.http.HttpHost;
import org.elasticsearch.action.bulk.BulkRequest;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.client.indices.GetIndexRequest;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import scala.Tuple2;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

/**
 * ES工具类， 用于对ES进行读写操作
 */
public class MyEsUtils {
    /**客户端对象*/
    public static RestHighLevelClient esClient = build();

    /**创建ES客户端对象*/
    private static RestHighLevelClient build() {
        String host = MyConfig.ES_HOST;
        String port = MyConfig.ES_PORT;
        RestClientBuilder restClientBuilder = RestClient.builder(new HttpHost(host,Integer.valueOf(port)));
        RestHighLevelClient client = new RestHighLevelClient(restClientBuilder);
        return  client;
    }

    /**关闭ES对象*/
    public static  void close() throws IOException {
        if(esClient != null) {
            esClient.close();
        }
    }

    /**
     * 1. 批写
     * 2. 幂等写
     */
    public static void bulkSave(String indexName, List<Tuple2<String,Object>>  docs) throws IOException {
        BulkRequest bulkRequest =  new BulkRequest(indexName);
        for(Tuple2<String,Object> doc: docs) {
            IndexRequest indexRequest = new IndexRequest();
            String dataJson = JSON.toJSONString(doc._2, new SerializeConfig(true));
            indexRequest.source(dataJson , XContentType.JSON);
            indexRequest.id(doc._1);
            bulkRequest.add(indexRequest);
        }
        esClient.bulk(bulkRequest , RequestOptions.DEFAULT);
    }

    /**
     * 查询指定的字段
     */
    public static List<String> searchField(String indexName, String fieldName) throws IOException {
        GetIndexRequest getIndexRequest = new GetIndexRequest(indexName);
        Boolean isExists = esClient.indices().exists(getIndexRequest,RequestOptions.DEFAULT);
        if(!isExists){
            return null;
        }
        //正常从ES中提取指定的字段
        List<String> mids = new ArrayList<>();
        SearchRequest searchRequest = new SearchRequest(indexName);
        SearchSourceBuilder searchSourceBuilder = new SearchSourceBuilder();
        searchSourceBuilder.fetchSource(fieldName,null).size(100000);
        searchRequest.source(searchSourceBuilder);
        SearchResponse searchResponse = esClient.search(searchRequest , RequestOptions.DEFAULT);
        SearchHit[] hits = searchResponse.getHits().getHits();
        for(SearchHit hit:hits) {
            Map<String,Object> sourceMap = hit.getSourceAsMap();
            String mid = sourceMap.get(fieldName).toString();
            mids.add(mid);
        }
        return  mids;
    }


}
